package com.cnd.scm.rabbitmq.producer;

import com.cnd.scm.rabbitmq.common.MessageSender;
import com.cnd.scm.rabbitmq.common.MessageSenderFactory;
import com.cnd.scm.rabbitmq.common.MessageStatusEnum;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @version V1.0
 * @ClassName: FastBuildRabbitMqProducer
 * @Description: 消息生产者
 * @author: 李启联
 * @date: 2021-02-24 11:00
 * @Copyright: Copyright (c), 2021
 */
@Component
public class FastBuildRabbitMqProducer{
    private static final Logger log = LoggerFactory.getLogger(FastBuildRabbitMqProducer.class);

    @Autowired
    private MessageSenderFactory messageSenderFactory;

    private ConnectionFactory connectionFactory;

    public FastBuildRabbitMqProducer(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }


    public MessageSender buildDirectMessageSender(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSender(exchange, routingKey, queue, "direct",null,false);
    }

    public MessageSender buildTopicMessageSender(final String exchange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, routingKey, null, "topic",null,false);
    }

    /**
     * 发送消息
     *
     * @param exchange   消息交换机
     * @param routingKey 消息路由key
     * @param queue      消息队列
     * @param type 消息队列类型(direct 直联模式 ; topic 主题模式)
     * @param arguments 附加参数
     * @param needDlx 是否创建对应的死信队列
     * return
     */
    public MessageSender buildMessageSender(final String exchange, final String routingKey, final String queue, final String type,Map<String,Object> arguments,boolean needDlx) throws IOException {
        Connection connection = connectionFactory.createConnection();

        if (type.equals("direct")) {
            buildQueue(exchange, routingKey, queue, connection, "direct",arguments,needDlx);
        } else if (type.equals("topic")) {
            buildQueue(exchange, routingKey, queue, connection, "topic",arguments,needDlx);
        }

        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        /**
         * 当mandatory标志位设置为true时
         * 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息
         * 那么broker会调用basic.return方法将消息返还给生产者
         * 当mandatory设置为false时，出现上述情况broker会直接将消息丢弃
         */
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setRoutingKey(routingKey);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        MessageSender messageSender = messageSenderFactory.createMessageSender();

//        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
//            String msgId = correlationData.getReturnedMessage().getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
//            if (!ack) {
//                log.info("send message failed: " + cause + correlationData.toString());
//                messageSender.updateStatus(MessageStatusEnum.FAIL,msgId);
//            }
//            else {
//                //retryCache.del(Long.valueOf(correlationData.getId()));
//                messageSender.updateStatus(MessageStatusEnum.SUCCESS,msgId);
//            }
//        });
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                String msgId = correlationData.getReturnedMessage().getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
                if (!ack) {
                    log.info("send message failed: " + cause + correlationData.toString());
                    //messageSender.updateStatus(MessageStatusEnum.FAIL,msgId);
                }else{
                    messageSender.updateStatus(MessageStatusEnum.SUCCESS,msgId);
                }
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                String msgId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
                log.info("send message failed: " + replyCode + " " + replyText);
                messageSender.updateStatus(MessageStatusEnum.FAIL,msgId);
            }
        });

        messageSender.setRabbitTemplate(rabbitTemplate);

        //messageSender.startRetry();
        return messageSender;

    }

    private void buildQueue(String exchange, String routingKey,final String queue, Connection connection, String type) throws IOException {
        buildQueue(exchange,routingKey,queue,connection,type,null,false);
    }

    /**
     * 构建一个消息队列
     * @param exchange 交换机名称
     * @param routingKey 路由键
     * @param queue 队列名称
     * @param connection 连接对象
     * @param type 消息队列类型(direct 直联模式 ; topic 主题模式)
     * @param arguments 附加参数
     * @param needDlx 是否创建对应的死信队列
     * @throws IOException
     */
    private void buildQueue(String exchange, String routingKey,
                            final String queue, Connection connection, String type, Map<String,Object> arguments, boolean needDlx) throws IOException {
        Channel channel = connection.createChannel(false);

        if (type.equals("direct")) {
            channel.exchangeDeclare(exchange, "direct", true, false, null);
        } else if (type.equals("topic")) {
            channel.exchangeDeclare(exchange, "topic", true, false, null);
        }
        //如果未指定队列名称,可以认定是无需要构建队列(构建topic类型的消息提供者交换机时，无需指定队列名称)
        if(null == queue || "".equals(queue)) {
            return;
        }

        if(needDlx) {
            channel.exchangeDeclare(exchange+".dlx", "direct", true, false, null);
            channel.queueDeclare(queue+".dlx", true, false, false, arguments);
            channel.queueBind(queue+".dlx", exchange+".dlx", queue+".routing.dlx");
            if(null == arguments){
                arguments = new HashMap<>(2);
            }
            arguments.put("x-dead-letter-exchange", queue+".dlx");
            arguments.put("x-dead-letter-routing-key", queue+".routing.dlx");
        }
        channel.queueDeclare(queue, true, false, false, arguments);

        channel.queueBind(queue, exchange, routingKey);
        try {
            channel.close();
        } catch (TimeoutException e) {
            log.info("close channel time out ", e);
        }
    }

    /**
     * 构建topic类型的消息提供者交换机
     * @param exchange 交换机名称
     * @param connection 连接对象
     * @throws IOException
     */
    private void buildTopic(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, "topic", true, false, null);
    }

}